package cn.edu360.report

import cn.edu360.beans.ReportAreaAnalysis
import cn.edu360.utils.{ConfigHandler, MySQLHandler, RptKpiTools}
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 地域分布 - core
  * sheep.Old @ 64341393
  * Created 2018/5/9
  */
object RptAreaAnalysis {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("地域分布")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)
    // 读取数据
    val rawDataFrame = sQLContext.read.parquet(ConfigHandler.parquetPath)
    import sQLContext.implicits._
    val result: DataFrame = rawDataFrame.map(row => {
      val pname = row.getAs[String]("provincename")
      val cname = row.getAs[String]("cityname")
      ((pname, cname), RptKpiTools.offLineKpi(row))
    })
      .reduceByKey((list1, list2) => list1 zip list2 map (tp => tp._1 + tp._2))
      .map(t => ReportAreaAnalysis(t._1._1, t._1._2, t._2(0), t._2(1), t._2(2), t._2(3), t._2(4), t._2(5), t._2(6), t._2(7), t._2(8)))
      .toDF()

    MySQLHandler.save2db(result, ConfigHandler.areaAnalysisTableName)
    sc.stop()
  }
}
